﻿#include "grpceventclient.h"
#include "grpceventdef.h"
#include "grpceventdispatcher.h"
#include "jsonwrapper.h"
#include <fstream>
#include <chrono>
#include <QDebug>

std::shared_ptr<GrpcEventClient> g_pGrpcClient = nullptr;
GrpcEventClient::GrpcEventClient(const std::string name, const std::string ip, const uint16_t port)
    : m_name(name)
    , m_ip(ip)
    , m_port(port)
{
    createConnection();

    m_thread = std::thread(std::bind(&GrpcEventClient::readWriteThreadProc, this));
}

GrpcEventClient::~GrpcEventClient()
{
    m_stop = true;
    if(m_thread.joinable())
    {
        m_thread.join();
    }
}

void GrpcEventClient::SendEvent(const std::string& dst, const std::string& eventName,
                                const nlohmann::json& array)
{
    if(!m_stop)
    {
        GrpcEventJsonAvis event;
        event.SourceId = m_name;
        event.TargetId = dst;
        event.EventName = eventName;
        //由于c# 那边是数组。。。
        event.Parameters = array;
        nlohmann::json json = event;

        GrpcEventMessage msg;
        msg.set_jsontext(json.dump());

        std::lock_guard<std::mutex> lk(m_queueMutex);
        m_eventQueue.push(msg);

    }
}

void GrpcEventClient::readWriteThreadProc()
{
    std::queue<GrpcEventMessage> sendQueue;
    while(!m_stop)
    {
        grpc_connectivity_state state = m_channel->GetState(true);
        switch(state)
        {
            case GRPC_CHANNEL_SHUTDOWN:
            {
                createConnection();
                std::this_thread::sleep_for(std::chrono::milliseconds(2000));
                break;
            }
            case GRPC_CHANNEL_READY:
            {
                if(sendQueue.empty())
                {
                    std::lock_guard<std::mutex> lk(m_queueMutex);
                    if(!m_eventQueue.empty())
                        std::swap(m_eventQueue, sendQueue);
                }

                //发送，由于服务器直接返回的，所以这里可以使用同步模式,一次最多连发5个。
                int cnt = 0;
                while(!sendQueue.empty() && cnt < 5)
                {
                    //注意这两个context ，response，对于阻塞式调用，不能使用全局或成员变量，会崩溃
                    ClientContext context;
                    GrpcEventMessage response;
                    //TODO 后续联调完可以不用打印
                    qDebug() << "Grpc Dispatch :" << sendQueue.front().jsontext().c_str();
                    Status status = m_stub->Dispatch(&context, sendQueue.front(), &response);
                    if (status.ok())
                    {
                        sendQueue.pop();
                        cnt++;
                    }
                    else
                    {
                        qCritical() << "Grpc Dispatch Error " << status.error_message().c_str();
                        if(!checkStatusCanRecover(status.error_code()))
                        {
                            qCritical() << "Grpc Status Can Not Recover Quit Thread";
                            m_stop = true ;
                        }
                        break;
                    }
                }

                //检查流状态
                if(m_reader->IsError())
                {
                    if(!checkStatusCanRecover(m_reader->GetErrStatus()))
                    {
                        qCritical() << "Grpc Status Can Not Recover Quit Thread";
                        m_stop = true ;
                    }
                    m_reader.reset();
                    m_reader = std::make_unique<Reader>(m_stub.get(), m_name);

                    m_reader->StartReadEvent();
                }

                std::this_thread::sleep_for(std::chrono::milliseconds(100));
                break;
            }
            default:
            {
                std::this_thread::sleep_for(std::chrono::milliseconds(1000));
                break;
            }
        }

    }
}

void GrpcEventClient::createConnection()
{
    if(m_channel)
    {
        m_channel.reset();
        m_stub.reset();
        qCritical() << "grpc client err and reconnect";
    }

    std::string server_address = absl::StrFormat("%s:%d", m_ip, m_port);

#ifdef TASK_SIMULATE
    m_channel = grpc::CreateChannel(server_address, grpc::InsecureChannelCredentials());
    m_stub = GrpcEvent::NewStub(m_channel);

    qInfo() << "CreateChannel server_address = " << server_address.c_str();

    m_reader.reset();
    m_reader = std::make_unique<Reader>(m_stub.get(), m_name);

    m_reader->StartReadEvent();
#else

    auto  ReadFile = [](const std::string& filename) {
        std::ifstream file(filename);
        if (!file.is_open()) {
            throw std::runtime_error("Unable to open file: " + filename);
        }
        return std::string((std::istreambuf_iterator<char>(file)), std::istreambuf_iterator<char>());
    };
    // 创建一个使用 HTTPS 的 gRPC 通道

    // 配置 SSL 证书
    grpc::SslCredentialsOptions ssl_opts;
    //ssl_opts.pem_root_certs = ReadFile("certificate.crt"); //  服务器证书
    ssl_opts.pem_root_certs = ReadFile("server.crt"); //  服务器证书
    auto creds = grpc::SslCredentials(ssl_opts);
    // 创建 SSL 凭据
    auto credentials = grpc::SslCredentials(ssl_opts);

    m_channel = grpc::CreateChannel(server_address, credentials);
#endif

}

bool GrpcEventClient::checkStatusCanRecover(grpc::StatusCode code)
{
    switch(code)
    {
        //如下两种重联也肯定会出错
    case grpc::StatusCode::UNAUTHENTICATED: //未认证
    case grpc::StatusCode::UNIMPLEMENTED: //服务器没有这个实现
    {
        return false;
    }
    default:
        return true;
    }
}



Reader::Reader(GrpcEvent::Stub* stub,  const std::string& parent)
    : m_parentStub(stub)
    , m_parent(parent)
{

}

void Reader::StartReadEvent()
{

    m_isError.store(false);

    GrpcEventMessage request;
    request.set_jsontext(m_parent);
    m_parentStub->experimental_async()->SubscribeDispatch(&m_context, &request, this);
    StartRead(&m_response);
    StartCall(); //启动前必须先启动读一个
}

bool Reader::IsError()
{
    return m_isError.load();
    return true;
}

grpc::StatusCode Reader::GetErrStatus()
{
    return m_errStatus;
}

void Reader::OnReadDone(bool ok)
{
    if (ok)
    {
        //分发
        std::pair<GrpcEventJsonAvis, bool> ret = ParseFromString<GrpcEventJsonAvis>(m_response.jsontext());
        if(ret.second)
        {
            if(m_parent != ret.first.TargetId)
            {
                qCritical() << "recv error event targetId = " << ret.first.TargetId.c_str();
            }
            else
            {
                g_pDispatcher->DispatchToLocal(ret.first.SourceId,  ret.first.EventName,  ret.first.Parameters);
            }
        }

        //读下一个
        StartRead(&m_response);
    }
    else
    {
        qCritical() << "Grpc Stream Read Error But Wait OnDone";
    }
}

void Reader::OnDone(const Status& s)
{
    if(s.ok())
    {
        qCritical() << "Grpc Stream Server Close ";
    }
    else
    {
        qCritical() << "Grpc Stream Error " << s.error_code();
    }

    m_errStatus = s.error_code();
    m_isError.store(true);
}
